import pymysql
import pandas as pd


def read_sql(sql, db_name):
    db_config = {
        "charset": "utf8",
        "host": "192.168.0.114",
        "port": 3307,
        "user": "admin",
        "passwd": "123456"
    }
    conn = pymysql.connect(db=db_name, **db_config)
    return pd.read_sql(sql, conn)


def df_into_db(df, db_name, table_name):
    db_config = {
        "charset": "utf8",
        "host": "192.168.0.114",
        "port": 3307,
        "user": "admin",
        "passwd": "123456"
    }
    columns_str = ",".join([f'`{col}`' for col in df.columns])
    place_holder = "%s,"*len(df.columns)
    sql = f"insert into {table_name} ({columns_str}) values ({place_holder[:-1]})"
    data = [tuple(x) for x in df.values]
    conn = pymysql.connect(db=db_name, **db_config)
    cursor = conn.cursor()
    cursor.executemany(sql, data)
    conn.commit()
    cursor.close()
    conn.close()
    print(f"{len(data)} rows inserted")


def get_tasks_detail(table_name):
    df = read_sql(f"select * from {table_name} ", db_name="task")
    df.drop_duplicates(subset=['script_name', 'func_name', 'args'], keep='first', inplace=True)
    df["project_name"] = table_name
    columns = ['script_name', 'func_name', 'args', 'project_name']
    df_into_db(df[columns], db_name="task", table_name="tasks_detail")


def get_execution_detail(table_name, date):
    task_details = read_sql(f"select * from tasks_detail where project_name = '{table_name}'", db_name="task")
    execution_df = read_sql(f"select * from {table_name} where date = '{date}'", db_name="task")
    execution_df.drop_duplicates(subset=['script_name', 'func_name', 'args', 'status'], keep='last', inplace=True)
    script_name_list = []
    func_name_list = []
    args_list = []
    is_success_list = []
    execution_time_list = []
    begin_time_list = []
    remark_list = []
    for _, row in task_details.iterrows():
        script_name = row['script_name']
        func_name = row['func_name']
        args = row['args']
        remark = row['remark']
        tmp_df = execution_df[(execution_df.script_name == script_name) & (execution_df.func_name == func_name) &
                              (execution_df.args == args)]
        is_success = False
        begin_time = ""
        if not tmp_df.empty and 1 in set(tmp_df["status"]):
            is_success = True
            begin_time = tmp_df.loc[tmp_df.status == 0, "created_at"].iloc[0]
            end_time = tmp_df.loc[tmp_df.status == 1, "updated_at"].iloc[0]
            execution_time = (end_time-begin_time).seconds
            begin_time = str(begin_time)
        else:
            execution_time = 0
        script_name_list.append(script_name)
        func_name_list.append(func_name)
        args_list.append(args)
        is_success_list.append(is_success)
        execution_time_list.append(execution_time)
        begin_time_list.append(begin_time)
        remark_list.append(remark)
    result_df = pd.DataFrame({
        "script_name": script_name_list,
        "func_name": func_name_list,
        "args": args_list,
        "is_success": is_success_list,
        "execution_time": execution_time_list,
        "begin_time": begin_time_list,
        "remark": remark_list
    })
    result_df.to_csv(f"{table_name}_execution_detail_{date}.csv", index=False)


if __name__ == "__main__":
    get_execution_detail("crypto_quant", "2025-09-14")
    get_execution_detail("crypto_quant", "2025-09-13")
    get_execution_detail("dataloader", "2025-09-14")
    get_execution_detail("dataloader", "2025-09-13")

